package engine

import (
	"context"
	"fmt"
	"regexp"
	"sort"
	"strings"
	"sync"

	"gitee.com/zfd81/dbrs/sql/meta"

	"gitee.com/zfd81/dbrs/sql/parser"

	"gitee.com/zfd81/dbrs/config"

	"gitee.com/zfd81/dbrs/expr"

	"vitess.io/vitess/go/mysql/collations"

	"github.com/spf13/cast"

	"github.com/go-redis/redis/v8"
	"vitess.io/vitess/go/sqltypes"
	querypb "vitess.io/vitess/go/vt/proto/query"
)

type RedisProvider struct {
	name   string
	typ    Type
	client redis.Cmdable
	mu     *sync.RWMutex
}

func (p *RedisProvider) Name() string {
	return p.name
}

func (p *RedisProvider) Type() Type {
	return p.typ
}

func (p *RedisProvider) Insert(ctx *Context, stmt *parser.InsertStatement) (*sqltypes.Result, error) {
	key, err := p.GetPrimaryKey(stmt.Table())
	if err != nil {
		return nil, err
	}
	keyIndex := -1
	for i, col := range stmt.Columns() {
		if key == col {
			keyIndex = i
			break
		}
	}
	if keyIndex < 0 {
		return nil, fmt.Errorf("Key %s not found", key)
	}

	rowkeyFunc := func(data []interface{}) string {
		return PrimaryKey(stmt.Table(), cast.ToString(data[keyIndex]))
	}

	affected, err := p.Set(rowkeyFunc, stmt.Columns(), stmt.Rows())
	return &sqltypes.Result{
		RowsAffected: affected,
	}, err
}

func (p *RedisProvider) Delete(ctx *Context, stmt *parser.DeleteStatement) (*sqltypes.Result, error) {
	key, err := p.GetPrimaryKey(stmt.Table())
	if err != nil {
		return nil, err
	}

	cols := stmt.Columns
	kvs := cols.Keyvalues(key)
	if kvs == nil || len(kvs) == 0 {
		return nil, fmt.Errorf("syntax error: primary key not found in where statement")
	}

	names := cols.Names()
	var keys []string
	for _, k := range kvs {
		primarykey := PrimaryKey(stmt.Table(), k)
		data, err := p.Get(ctx, primarykey, names...)
		if err != nil {
			return nil, err
		}
		if data != nil && len(data) > 0 {
			ok, err := stmt.Where().Eval(data)
			if err != nil {
				return nil, err
			}
			if ok {
				keys = append(keys, primarykey)
			}
		}
	}

	var affected int64
	if len(keys) > 0 {
		affected, err = p.Remove(keys...)
		if err != nil {
			return nil, err
		}
	}
	return &sqltypes.Result{
		RowsAffected: uint64(affected),
	}, nil
}

func (p *RedisProvider) Update(ctx *Context, stmt *parser.UpdateStatement) (*sqltypes.Result, error) {
	key, err := p.GetPrimaryKey(stmt.Table())
	if err != nil {
		return nil, err
	}

	cols := stmt.Columns
	kvs := cols.Keyvalues(key)
	if kvs == nil || len(kvs) == 0 {
		return nil, fmt.Errorf("syntax error: primary key not found in where statement")
	}

	names := cols.Names()
	var rows [][]interface{}
	for _, k := range kvs {
		primarykey := PrimaryKey(stmt.Table(), k)
		data, err := p.Get(ctx, primarykey, names...)
		if err != nil {
			return nil, err
		}
		if data != nil && len(data) > 0 {
			ok, err := stmt.Where().Eval(data)
			if err != nil {
				return nil, err
			}
			if ok {
				vm := expr.BuildVM(data)
				row := make([]interface{}, len(stmt.Fields)+1)
				row[0] = k
				for index, field := range stmt.Fields {
					v, err := vm.Eval(field.Expr())
					if err != nil {
						return nil, err
					}
					row[index+1] = v
				}
				rows = append(rows, row)
			}
		}
	}

	rowkeyFunc := func(data []interface{}) string {
		return PrimaryKey(stmt.Table(), cast.ToString(data[0]))
	}
	attrs := make([]string, len(stmt.Fields)+1)
	attrs[0] = key
	for i, field := range stmt.Fields {
		attrs[i+1] = field.Name()
	}
	affected, err := p.Set(rowkeyFunc, attrs, rows)
	return &sqltypes.Result{
		RowsAffected: affected,
	}, err
}

func (p *RedisProvider) Select(ctx *Context, bindVars map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error, stmt *parser.SelectStatement) error {
	key, err := p.GetPrimaryKey(stmt.Table())
	if err != nil {
		return err
	}

	cols := stmt.Columns
	kvs := cols.Keyvalues(key)
	if kvs == nil || len(kvs) == 0 {
		return fmt.Errorf("syntax error: primary key not found in where statement")
	}

	r := &sqltypes.Result{
		Fields: []*querypb.Field{},
		Rows:   [][]sqltypes.Value{},
	}

	names := cols.Names()
	for _, field := range stmt.Fields {
		r.Fields = append(r.Fields, &querypb.Field{
			Name:    field.Name(),
			Type:    sqltypes.VarChar,
			Charset: collations.CollationUtf8ID,
		})
	}
	for _, k := range kvs {
		primarykey := PrimaryKey(stmt.Table(), k)
		data, err := p.Get(ctx, primarykey, names...)
		if err != nil {
			return err
		}
		if data != nil && len(data) > 0 {
			row := []sqltypes.Value{}
			ok, err := stmt.Where().Eval(data)
			if err != nil {
				return err
			}
			if ok {
				vm := expr.BuildVM(data)
				for _, field := range stmt.Fields {
					v, err := vm.Eval(field.Expr())
					if err != nil {
						return err
					}
					typ, val := sqlTypes(v)
					row = append(row, sqltypes.MakeTrusted(typ, []byte(val)))
				}
				r.Rows = append(r.Rows, row)
			}
		}
	}
	r.RowsAffected = uint64(len(r.Rows))
	callback(r)
	return nil
}

func (p *RedisProvider) ShowTables(ctx *Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	rctx := context.Background()
	keys, err := p.client.Keys(rctx, fmt.Sprintf("%s*", TablePrefix(meta.TABLES))).Result()
	if err != nil {
		return nil, err
	}
	r := &sqltypes.Result{}
	schema := stmt.Node().Schema()
	fields := schema.ToFields()
	r.Fields = fields
	reg := "^" + strings.ReplaceAll(stmt.Filter, "%", "\\w*") + "$"
	for _, key := range keys {
		name := key[6:]
		matched, _ := regexp.MatchString(reg, name)
		if stmt.Filter == "" || matched {
			vals, err := p.client.HMGet(ctx, key, meta.TABLES_TABLE_NAME, meta.TABLES_ENGINE, meta.TABLES_PRIMARY_KEY).Result()
			if err != nil {
				return nil, err
			}
			row := schema.CreateRow()
			row.Append(vals[0])
			if stmt.Full {
				row.Append(vals[1])
			}
			row.Append(fmt.Sprintf("PrimaryKey: %s", vals[2]))
			row.Append("")
			r.Rows = append(r.Rows, row.ToSQL())
		}
	}
	r.RowsAffected = uint64(len(r.Rows))
	return r, nil
}

func (p *RedisProvider) ShowCharset(ctx *Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Currently redis engine does not support 'show charset' statement")
}
func (p *RedisProvider) ShowCollation(ctx *Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Currently redis engine does not support 'show collation' statement")
}

func (p *RedisProvider) DescTable(ctx *Context, stmt *parser.DescStatement) (*sqltypes.Result, error) {
	tbl := stmt.Tbl()
	key, err := p.GetPrimaryKey(tbl)
	if err != nil {
		return nil, fmt.Errorf("Table '%s' doesn't exist", tbl)
	}

	columns := map[string]interface{}{}
	var cursor uint64
	for i := 0; i < 5; i++ {
		var keys []string
		var err error
		keys, cursor, err = p.client.Scan(ctx, cursor, fmt.Sprintf("%s*", TablePrefix(tbl)), 100000).Result()
		if err != nil {
			return nil, err
		}
		if cursor == 0 { // no more keys
			break
		}
		kvs, err := p.client.HGetAll(ctx, keys[0]).Result()
		if err != nil {
			return nil, err
		}
		for k, v := range kvs {
			columns[k] = v
		}
		cursor = cursor + 700000
	}
	r := &sqltypes.Result{}
	schema := stmt.Node().Schema()
	fields := schema.ToFields()
	r.Fields = fields
	cols := []string{}
	for k, _ := range columns {
		cols = append(cols, k)
	}
	sort.Sort(sort.StringSlice(cols))
	for _, col := range cols {
		row := schema.CreateRow()
		row.Append(col)
		typ, val := sqlTypes(columns[col])
		row.Append(typ)
		row.Append(true)
		if col == key {
			row.Append("PRI")
		} else {
			row.Append("")
		}
		row.Append("")
		row.Append("")
		row.Append(fmt.Sprintf("eg: %s", val))
		r.Rows = append(r.Rows, row.ToSQL())
	}
	r.RowsAffected = uint64(len(r.Rows))
	return r, nil
}

func (p *RedisProvider) AlterTable(ctx *Context, stmt *parser.AlterTableStatement) (*sqltypes.Result, error) {
	if stmt.PrimaryKey == "" {
		return nil, fmt.Errorf("Primary key cannot be empty")
	}
	_, err := p.client.HSet(context.Background(), PrimaryKey(meta.TABLES, stmt.Tbl), meta.TABLES_TABLE_NAME, stmt.Tbl, meta.TABLES_ENGINE, "Redis", meta.TABLES_PRIMARY_KEY, stmt.PrimaryKey).Result()
	if err != nil {
		return nil, err
	}
	return &sqltypes.Result{
		RowsAffected: 1,
	}, nil
}

func (p *RedisProvider) DropTable(ctx *Context, stmt *parser.DropTableStatement) (*sqltypes.Result, error) {
	rctx := context.Background()
	var affected int64 = 0
	for _, tbl := range stmt.Tbls {
		cnt, err := p.client.Del(rctx, PrimaryKey(meta.TABLES, tbl)).Result()
		if err != nil {
			return nil, err
		}
		if cnt > 0 {
			var cursor uint64
			for {
				var keys []string
				var err error
				keys, cursor, err = p.client.Scan(rctx, cursor, fmt.Sprintf("%s*", TablePrefix(tbl)), 5000).Result()
				if err != nil {
					return nil, err
				}
				if len(keys) > 0 {
					_, err = p.client.Del(ctx, keys...).Result()
					if err != nil {
						return nil, err
					}
				}
				if cursor == 0 { // no more keys
					break
				}
			}
			affected = affected + cnt
		}
	}
	return &sqltypes.Result{
		RowsAffected: uint64(affected),
	}, nil
}

func (p *RedisProvider) Set(rowkeyFunc RowkeyFunc, attrs []string, rows [][]interface{}) (uint64, error) {
	ctx := context.Background()
	cmds, err := p.client.Pipelined(ctx, func(pipe redis.Pipeliner) error {
		for _, row := range rows {
			pipe.HSet(ctx, rowkeyFunc(row), rowFunc(attrs, row))
		}
		return nil
	})
	return uint64(len(cmds)), err
}

func (p *RedisProvider) Get(ctxt *Context, key string, attrs ...string) (map[string]interface{}, error) {
	result := map[string]interface{}{}
	size := len(attrs)
	ctx := context.Background()
	if size > 0 {
		empty := 0
		vals, err := p.client.HMGet(ctx, key, attrs...).Result()
		if err != nil {
			return result, err
		}
		for i, attr := range attrs {
			if vals[i] == nil {
				empty++
			}
			result[attr] = vals[i]
		}
		if empty == size {
			return nil, nil
		}
	} else {
		kvs, err := p.client.HGetAll(ctx, key).Result()
		if err != nil {
			return result, err
		}
		for k, v := range kvs {
			result[k] = v
		}
	}
	for k, v := range ctxt.GetAllEnvironmentVariables() {
		result[k] = v
	}
	return result, nil
}

func (p *RedisProvider) Remove(keys ...string) (int64, error) {
	return p.client.Del(context.Background(), keys...).Result()
}

//func (p *RedisProvider) getMetadata(table string, attrs ...string) ([]interface{}, error) {
//	return p.client.HMGet(context.Background(), PrimaryKey(meta.TABLES, table), attrs...).Result()
//}

func (p *RedisProvider) GetPrimaryKey(table string) (string, error) {
	vals, err := p.client.HMGet(context.Background(), PrimaryKey(meta.TABLES, table), meta.TABLES_PRIMARY_KEY).Result()
	if err != nil {
		return "", err
	}
	if vals[0] == nil {
		return "nil", fmt.Errorf("Primary key not found for table %s", table)
	}
	return cast.ToString(vals[0]), nil
}

func (p *RedisProvider) Client() redis.Cmdable {
	return p.client
}

func NewRedisProvider(db config.Database) (*RedisProvider, error) {
	var rdb redis.Cmdable
	if len(db.Endpoints) == 1 {
		opts := &redis.Options{
			Addr:     db.Endpoints[0],
			Password: db.Password,
			DB:       cast.ToInt("0"),
		}
		rdb = redis.NewClient(opts)
	} else {
		opts := &redis.ClusterOptions{
			Addrs:    db.Endpoints,
			Password: db.Password,
		}
		rdb = redis.NewClusterClient(opts)
	}
	return &RedisProvider{
		name:   db.Name,
		typ:    Type_Redis,
		client: rdb,
		mu:     &sync.RWMutex{},
	}, nil
}

type RowkeyFunc func(data []interface{}) string

func rowFunc(attrs []string, datas []interface{}) []interface{} {
	values := make([]interface{}, len(attrs)*2)
	for i, name := range attrs {
		values[i*2] = name
		values[i*2+1] = datas[i]
	}
	return values
}

func PrimaryKey(tbl, key string) string {
	return "$" + tbl + "_" + key
}

func TablePrefix(tbl string) string {
	return "$" + tbl + "_"
}
